package com.xiashitech.agent.grpc;

import com.xiashitech.agent.cache.status.CacheStatus;
import com.xiashitech.agent.config.StartConfig;
import com.xiashitech.agent.constants.ErrorCode;
import com.xiashitech.agent.utils.JsonUtil;
import com.xiashitech.interfaces.agent.dto.SystemInfoDTO;
import com.xiashitech.interfaces.apiservice.dto.system.SystemInfoRequest;
import com.xiashitech.interfaces.apiservice.dto.system.SystemInfoResponse;
import com.xiashitech.interfaces.grpc.RPCAgentStatusServiceGrpc;
import com.xiashitech.interfaces.grpc.RPCSystemInfoRequest;
import com.xiashitech.interfaces.grpc.RPCSystemInfoResponse;
import io.grpc.Deadline;
import io.grpc.ManagedChannel;
import io.grpc.stub.StreamObserver;
import io.opentelemetry.javaagent.bootstrap.internal.InstrumentationConfig;

import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class XsAgentStatusClient {
    private final static XsAgentStatusClient client = new XsAgentStatusClient();
    private volatile StreamObserver<RPCSystemInfoRequest> soAgentHeartBeat = null;
    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    private RPCAgentStatusServiceGrpc.RPCAgentStatusServiceStub agentStub = null;
    private volatile boolean apiStreamError = false;

    public static XsAgentStatusClient getClient() {
        return client;
    }

    /**
     * 本方法阻塞，线程安全
     * 当前为在应用启动时，单次调用
     * @param param
     * @return
     */
    public String createAgentStatus(String param) {
        String respStr = null;
        try {
            System.out.println(Thread.currentThread().getId() + ", at: " + System.currentTimeMillis() + ", createAgentStatus enter, StartConfig.agentService: " + StartConfig.getParamValueFromKey(StartConfig.agentService));
            ManagedChannel channelRef = XsChannelManager.getManager().buildStubOfChannel("127.0.0.1", 9988);
            RPCAgentStatusServiceGrpc.RPCAgentStatusServiceBlockingStub blockingStub = RPCAgentStatusServiceGrpc.newBlockingStub(channelRef);
            RPCSystemInfoRequest request = RPCSystemInfoRequest.newBuilder().setSystemInfoRequestJson(param).build();
            System.out.println(Thread.currentThread().getId() + ", at: " + System.currentTimeMillis() + ", createAgentStatus start blocking grpc, para: " + param);
            RPCSystemInfoResponse response = blockingStub.grpcRegistAgentStatus(request);
            respStr = response.getSystemInfoResponseJson();
            System.out.println(Thread.currentThread().getId() + ", at: " + System.currentTimeMillis() + ", createAgentStatus finished, return string: " + respStr);
        } catch (Exception e) {
            StartConfig.throwableStackPrinter(e);
        }
        return respStr;
    }

    /**
     * 本方法阻塞调用，线程安全
     * 不宜在主线程中调用，会阻塞主线程
     * 阻塞时长为 5s
     * 当前可能会多线程调用，应该并发度不高
     * @param param
     * @return
     */
    public String createAgentHost(String param) {
        String respStr = null;
        try {
            System.out.println(Thread.currentThread().getId() + ", createAgentStatus enter, StartConfig.agentService: " + StartConfig.getParamValueFromKey(StartConfig.agentService));
            ManagedChannel channelRef = XsChannelManager.getManager().buildStubOfChannel("192.168.0.54", 9988);
            RPCAgentStatusServiceGrpc.RPCAgentStatusServiceBlockingStub blockingStub = RPCAgentStatusServiceGrpc.newBlockingStub(channelRef).withDeadline(Deadline.after(5, TimeUnit.SECONDS));
            RPCSystemInfoRequest request = RPCSystemInfoRequest.newBuilder().setSystemInfoRequestJson(param).build();
            RPCSystemInfoResponse response = blockingStub.grpcSyncAgentHost(request);
            respStr = response.getSystemInfoResponseJson();
        } catch (Exception e) {
            StartConfig.throwableStackPrinter(e);
        }
        return respStr;
    }

    /**
     * 本方法为线程安全，当前调用是单线程长跨度调用
     * 本方法为异步调用
     * 但：
     *   多次调用本方法，表示整个通道已重建，soAgentHeartBeat 也要重新创建
     *   如果 soAgentHeartBeat 不重建，无法维持通道
     *   因此，如果不是通道重建，不要多次调用本方法，更不要并发调用本方法
     * @param params
     */
    public void agentStatus(Map<String, String> params) {
        System.out.println(Thread.currentThread().getId() + ", agentStatus enter, StartConfig.agentService: " + StartConfig.getParamValueFromKey(StartConfig.agentService));
        ManagedChannel channelRef = XsChannelManager.getManager().buildStubOfChannel("127.0.0.1", 9988);
        System.out.println(Thread.currentThread().getId() + ", agentStatus now channel identityHashCode: "+ System.identityHashCode(channelRef));
        SystemInfoRequest systemInfoRequest = SystemInfoRequest.builder().id(params.get("systemId")).microServiceId(params.get("microServiceId")).build();
        RPCSystemInfoRequest rpcSystemInfoRequest = RPCSystemInfoRequest.newBuilder().setSystemInfoRequestJson(JsonUtil.convertObjectToString(systemInfoRequest)).build();
        System.out.println(Thread.currentThread().getId() + ", agentStatus systemInfoRequest is: " + systemInfoRequest.getId()+", "+systemInfoRequest.getMicroServiceId());

        synchronized (client) { // 主要是用于看 soAgentHeartBeat 是否 null
            if(agentStub == null ) {
                // 构建响应对象
                final StreamObserver<RPCSystemInfoResponse> soResponse = new StreamObserver<RPCSystemInfoResponse>() {
                    AtomicInteger cnt = new AtomicInteger(0);
                    @Override
                    public void onNext(RPCSystemInfoResponse value) {
                        System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() + " receive agentStatus times: " + cnt.incrementAndGet());
                        System.out.println(Thread.currentThread().getId() + ", agentStatus response value is: \n" + value);
                        String response = value.getSystemInfoResponseJson();
                        if("AGENTHEARTBEAT".equals(response)) {
                            System.out.println(Thread.currentThread().getId() + " receive agentStatus heartbeat response, skip");
                            return;
                        }
                        System.out.println(Thread.currentThread().getId() + " receive pushed agentStatus INFO ........ ");
                        if (response != null && !response.isEmpty() && JsonUtil.isJson(response)) {
                            SystemInfoResponse systemInfoResponse = JsonUtil.convertStringToObject(response, SystemInfoResponse.class);
                            if (systemInfoResponse != null && ErrorCode.RESULT_1001.getCode().equals(systemInfoResponse.getCode())) {
                                List<SystemInfoDTO> systemInfos = systemInfoResponse.getSystemInfos();
                                if (systemInfos != null) {
                                    // 以下方法可能线程不安全。实际业务中，应该服务端的推送的并发度基本不存在，这里可不用考虑并发问题
                                    CacheStatus.setCacheStatusByValue(systemInfos.get(0).getAgentStatus());
                                }
                            }
                        }
                    }

                    @Override
                    public void onError(Throwable t) {
                        System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() +", agentStatus onError the error: " + t.getMessage());
                        // XsChannelManager.getManager().reportError(channelRef_local);
                        apiStreamError = true;
                    }

                    @Override
                    public void onCompleted() {
                        System.out.println(Thread.currentThread().getId() + ", agentStatus Response onCompleted. EXCEPTION ？");
                        soAgentHeartBeat.onCompleted();
                        soAgentHeartBeat = null;
                    }

                    @Override
                    protected void finalize() throws Throwable { // 看看是否会被清理
                        System.out.println("StreamObserver<RPCSystemInfoResponse> is be GC: " + this.hashCode());
                        super.finalize();
                    }
                };

                long period = 50; // 50 秒
                if(!StartConfig.isProxyGrpc()) {
                    period = 600; // 10 分钟
                }
                // 构建定时器
                scheduledExecutorService.scheduleAtFixedRate(() -> {
                    try {
                        Thread.currentThread().setName("XsAgentStatusHeartBeat");
                        if(agentStub == null ) {
                            System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() + ", agentStub is null, EXCEPTION！！！");
                            return;
                        }
                        String requestStr = "AGENTHEARTBEAT";
                        // 可能流已经坏了，需要重建流
                        if(apiStreamError || soAgentHeartBeat == null) {
                            apiStreamError = false;
                            // 当流坏了要重建时，需要重新请求，以创建服务端的推送端点
                            String systemId = StartConfig.getParamValueFromKey(StartConfig.systemId);
                            String microServiceId = InstrumentationConfig.get().getString("otel.service.name");
                            SystemInfoRequest sysInfoRequest = SystemInfoRequest.builder().id(systemId).microServiceId(microServiceId).build();
                            requestStr = JsonUtil.convertObjectToString(sysInfoRequest);
                            // 重建 stub
                            System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() + ", apiStreamError is false, need rebuild request StreamObserver");
                            synchronized (client) { // 主要是用于看 soAgentHeartBeat 是否 null
                                StreamObserver<RPCSystemInfoRequest> soSIRequest = agentStub.biAgentStatus(soResponse); // 采用双向流，定时发送请求，表示心跳
                                soAgentHeartBeat = soSIRequest;
                            }
                        }
                        if(soAgentHeartBeat == null) {
                            System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() + ", soAgentHeartBeat is null, ************* SHOULD NOT");
                            return;
                        }

                        System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() + ", request a AgentStatus HeartBeart");
                        RPCSystemInfoRequest rpcSysInfoRequest = RPCSystemInfoRequest.newBuilder().setSystemInfoRequestJson(requestStr).build();
                        synchronized (soAgentHeartBeat) { // 应该不会存在对 soAgentHeartBeat 的并发调用
                            System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() + ", agentStatus onNext+++++++++++ soAgentHeartBeat: " + System.identityHashCode(soAgentHeartBeat));
                            soAgentHeartBeat.onNext(rpcSysInfoRequest);
                        }
                    } catch (Throwable t) {
                        StartConfig.throwableStackPrinter(t);
                    }
                }, 1, period, TimeUnit.SECONDS);

                // 构建成员缓存对象
                agentStub = RPCAgentStatusServiceGrpc.newStub(channelRef);
                StreamObserver<RPCSystemInfoRequest> soSystemInfoRequest = agentStub.biAgentStatus(soResponse); // 采用双向流，定时发送请求，表示心跳
                soAgentHeartBeat = soSystemInfoRequest;
            }
        }
        synchronized (soAgentHeartBeat) {
            System.out.println(Thread.currentThread().getId() + " T" + System.nanoTime() +", agentStatus onNext---------- soAgentHeartBeat: " + System.identityHashCode(soAgentHeartBeat));
            soAgentHeartBeat.onNext(rpcSystemInfoRequest);
        }
    }
}
